package com.tmt.reactor.mono;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import reactor.core.publisher.Mono;

/**
 * mono 测试
 * 
 * Mono<Void>: 如果第一次得到的是 Mono<Void> 则直接会执行 onComplete 或 onError， 
 *             但会执行后面的 doOnSuccess
 * Mono<Object> : 中间是不会传递 null， 如果某个环节 数据为 null 则直接 onComplete 或 onError
 * 
 * 每个环节不能返回 null 对象
 * @author lifeng
 */
public class MonoTestMain {

	/**
	 * 如果是 Void 返回 null是可以走到最底层的
	 */
	public static void void_future() {
		CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
             System.out.println("异步任务");
		});
		Mono<Void> mono = Mono.fromCompletionStage(future);
		mono.map((v) ->{
			System.out.println("进一步处理1");
			return "123";
		}).map((v) ->{
			System.out.println("进一步处理2");
			return null;
		}).doOnSuccess((v) ->{
			System.out.println("doOnSuccess");
		}).subscribe(new Subscriber<Object>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(Long.MAX_VALUE);
			}
			@Override
			public void onNext(Object t) {
				System.out.println("onNext");
			}
			@Override
			public void onError(Throwable t) {
				System.out.println("onError");
			}
			@Override
			public void onComplete() {
				System.out.println("onComplete");
			}
		});
	}
	
	/**
	 * object 中间是不能传递null 的，如果强制传递 null 则直接抛出异常
	 */
	public static void object_future() {
		
		CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
             System.out.println("异步任务");
             return "数据";
		});
		
		Mono<Object> mono = Mono.fromCompletionStage(future);
		mono.map((v) ->{
			System.out.println("进一步处理1");
			return v;
		}).map((v) ->{
			System.out.println("进一步处理2");
			return null;
		}).doOnSuccess((v) ->{
			System.out.println("doOnSuccess");
		}).subscribe(new Subscriber<Object>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(Long.MAX_VALUE);
			}
			@Override
			public void onNext(Object t) {
				System.out.println("onNext");
			}
			@Override
			public void onError(Throwable t) {
				System.out.println("onError");
			}
			@Override
			public void onComplete() {
				System.out.println("onComplete");
			}
		});
	}

	public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        object_future();
        latch.await();
	}
}